package biz.datalk.industrialland.admin.config;

import biz.datalk.industrialland.admin.config.properties.RedisDelayTaskExecutorProperties;
import biz.datalk.industrialland.admin.config.properties.TaskScheduleExecutorProperties;
import biz.datalk.industrialland.common.component.filter.MDCFilterTaskDecorator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author tarofang@163.com
 * @date 2020-06-24
 */
@Slf4j
@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties({ //
        TaskScheduleExecutorProperties.class //
        , RedisDelayTaskExecutorProperties.class //
})
public class ThreadPoolExecutorConfig {
    public static final String EXECUTOR_BEAN_NAME_TASK_SCHEDULE = "taskScheduleThreadPoolTaskExecutor";
    public static final String EXECUTOR_BEAN_NAME_REDIS_DELAY_TASK = "redisDelayTaskThreadPoolTaskExecutor";
    public static final String SEMAPHORE_BEAN_NAME_REDIS_DELAY_TASK = "redisDelayTaskSemaphore";

    /**
     * 定时任务执行的线程池
     * 不会将 RequestAttributes 传递到线程中
     */
    @Bean(EXECUTOR_BEAN_NAME_TASK_SCHEDULE)
    public ThreadPoolTaskExecutor taskScheduleThreadPoolTaskExecutor(TaskScheduleExecutorProperties taskScheduleExecutorProperties) {
        log.debug("init taskScheduleThreadPoolTaskExecutor ===========> start");
        try {
            ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
            threadPoolTaskExecutor.setCorePoolSize(taskScheduleExecutorProperties.getCorePoolSize());
            threadPoolTaskExecutor.setMaxPoolSize(taskScheduleExecutorProperties.getMaxPoolSize());
            threadPoolTaskExecutor.setKeepAliveSeconds(taskScheduleExecutorProperties.getKeepAliveSeconds());
            threadPoolTaskExecutor.setQueueCapacity(taskScheduleExecutorProperties.getQueueCapacity());
            threadPoolTaskExecutor.setThreadNamePrefix(taskScheduleExecutorProperties.getThreadNamePrefix());
            threadPoolTaskExecutor.setAllowCoreThreadTimeOut(false);
            threadPoolTaskExecutor.setDaemon(false);
            // 线程优先级
            threadPoolTaskExecutor.setThreadPriority(Thread.NORM_PRIORITY);
            threadPoolTaskExecutor.setWaitForTasksToCompleteOnShutdown(taskScheduleExecutorProperties.isWaitForTasksToCompleteOnShutdown());
            threadPoolTaskExecutor.setAwaitTerminationSeconds(taskScheduleExecutorProperties.getAwaitTerminationSeconds());

            // setRejectedExecutionHandler：当pool已经达到max size的时候，如何处理新任务
            // CallerRunsPolicy：不在新线程中执行任务，而是由调用者所在的线程来执行
            threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

            threadPoolTaskExecutor.setTaskDecorator(new MDCFilterTaskDecorator(taskScheduleExecutorProperties.getTransferRequestAttributes()));

            threadPoolTaskExecutor.afterPropertiesSet();
            return threadPoolTaskExecutor;
        } finally {
            log.debug("init taskScheduleThreadPoolTaskExecutor ===========> finish");
        }
    }

    /**
     * 同步外部联系人 Semaphore
     */
    @Bean("taskScheduleSemaphore")
    public Semaphore taskScheduleSemaphore(TaskScheduleExecutorProperties taskScheduleExecutorProperties) {
        log.debug("init taskScheduleSemaphore ===========> start");
        try {
            return new Semaphore(taskScheduleExecutorProperties.getSemaphoreNum());
        } finally {
            log.debug("init taskScheduleSemaphore ===========> finish");
        }
    }

    @Bean(EXECUTOR_BEAN_NAME_REDIS_DELAY_TASK)
    @ConditionalOnProperty(name = "executor.redis-delay-task.enable", havingValue = "true", matchIfMissing = true)
    public ThreadPoolTaskExecutor redisDelayTaskThreadPoolTaskExecutor(RedisDelayTaskExecutorProperties redisDelayTaskExecutorProperties) {
        log.debug("init redisDelayTaskThreadPoolTaskExecutor ===========> start");
        try {
            ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
            threadPoolTaskExecutor.setCorePoolSize(redisDelayTaskExecutorProperties.getCorePoolSize());
            threadPoolTaskExecutor.setMaxPoolSize(redisDelayTaskExecutorProperties.getMaxPoolSize());
            threadPoolTaskExecutor.setKeepAliveSeconds(redisDelayTaskExecutorProperties.getKeepAliveSeconds());
            threadPoolTaskExecutor.setQueueCapacity(redisDelayTaskExecutorProperties.getQueueCapacity());
            threadPoolTaskExecutor.setThreadNamePrefix(redisDelayTaskExecutorProperties.getThreadNamePrefix());
            threadPoolTaskExecutor.setAllowCoreThreadTimeOut(false);
            threadPoolTaskExecutor.setDaemon(false);
            // 线程优先级
            threadPoolTaskExecutor.setThreadPriority(Thread.NORM_PRIORITY);
            threadPoolTaskExecutor.setWaitForTasksToCompleteOnShutdown(redisDelayTaskExecutorProperties.isWaitForTasksToCompleteOnShutdown());
            threadPoolTaskExecutor.setAwaitTerminationSeconds(redisDelayTaskExecutorProperties.getAwaitTerminationSeconds());

            // setRejectedExecutionHandler：当pool已经达到max size的时候，如何处理新任务
            // CallerRunsPolicy：不在新线程中执行任务，而是由调用者所在的线程来执行
            threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

            threadPoolTaskExecutor.setTaskDecorator(new MDCFilterTaskDecorator(redisDelayTaskExecutorProperties.getTransferRequestAttributes()));

            threadPoolTaskExecutor.afterPropertiesSet();
            return threadPoolTaskExecutor;
        } finally {
            log.debug("init redisDelayTaskThreadPoolTaskExecutor ===========> finish");
        }
    }

    @Bean(SEMAPHORE_BEAN_NAME_REDIS_DELAY_TASK)
    @ConditionalOnProperty(name = "executor.redis-delay-task.enable", havingValue = "true", matchIfMissing = true)
    public Semaphore redisDelayTaskSemaphore(RedisDelayTaskExecutorProperties redisDelayTaskExecutorProperties) {
        log.debug("init redisDelayTaskSemaphore ===========> start");
        try {
            return new Semaphore(redisDelayTaskExecutorProperties.getSemaphoreNum());
        } finally {
            log.debug("init redisDelayTaskSemaphore ===========> finish");
        }
    }

}
